/* Not running / tested*/

#include "bullyalgo.h"

long cycles[USERS], sent[USERS], cycles_bully, bytes_sent, bytes_recvd;
clock_t st_time, curr_time;
FILE *f;
char filenam[20];
int electn = 0, leader = -1;
long double tm_lastmsg[USERS];

//Input the node of the talker when calling this function
void start_bullyalgo(int argc, char *argv[])
{
  int ret3, ret4, ret5, ret6;
  pthread_t appthsend, appthrecv, appsendrecv, thread_time;
  if(argc!=2)
    {
      fprintf(stderr, "usage: talker\'s_node_number\n");
      exit(1);
    }
  fopen_bully(argv[1]);

  if((ret5=pthread_create( &appsendrecv, NULL, &main_sendrecv, (void *)argv[1])))
   {
      	printf("Thread creation failed: %d\n", ret5);
   }
  t_virt=t_init=st_time = clock();
  if((ret6=pthread_create( &thread_time, NULL, &time_virtual, NULL)))
   {
      	printf("Thread creation failed: %d\n", ret6);
   }

  sleep(2);
  if((ret3=pthread_create( &appthsend, NULL, &appsendbully, NULL)))
   {
      	printf("Thread creation failed: %d\n", ret3);
   }

  if((ret4=pthread_create( &appthrecv, NULL, &apprecvbully, NULL)))
   {
        printf("Thread creation failed: %d\n", ret4);
   }
   pthread_join(appthsend, NULL);
   pthread_join(appthrecv, NULL);
   close(sockfd);
   return;
}

void fopen_bully(char *nodeno)
{
  strcpy(filenam, nodeno);
  strcat(filenam, "electn");
  if((f=fopen(filenam, "w"))==NULL)
    {
      printf("Can't open file to write\n");
      exit(1);
    }
}

//Application before sending to queue
void *appsendbully()
{
  int i;
  unsigned long j;
  msg_clksync *msg;
  struct sockaddr_in dest_add;
  for(i=0;i<USERS;i++)
    sent[i]=0;
  for(j=0;1;j++)
    {
      sleep(DT);
      for(i=0;i<USERS;i++)
	{
	  if(i==hno)
	    continue;
	  msg = (msg_clksync *)malloc(sizeof(msg_clksync));
	  msg->type = SEND_SYNC1;
	  msg->t0=tm_virt + tm_clock - tm_init;
	  msg->machine=i;
	  msg->iteratn=j;
	  dest_add = addr[i];
	  queue_send(msg, sizeof(msg_clksync), dest_add);
	  sent[i]++;
	  //printf("Sent %d, %lu\n", i, j);
	  free(msg);
	}
    }
  return (NULL);
}

//Application for receiving and working with the receives
void *apprecvbully()
{
  long double tstrt, tsnd, trcv, tend, max_time, max_wait_time;
  int i, size_msg, k, recvd;
  char *msg_ch;
  unsigned long j;
  msg_s *mw;
  struct sockaddr_in their_addr;
  long double t_diff, t_max[USERS], t_min[USERS], t_avg[USERS];
  for(i=0;i<USERS;i++)
    {
      t_max[i]=0.0;
      t_min[i]=999999.0;
      t_avg[i]=0.0;
      cycles[i]=0;
      tm_lastmsg[i] = 0.0;
    }
  cycles_bully=0;
  while(1)
    {
      msg_ch = (char *)malloc(sizeof(msg_s) + sizeof(char));
      if((queue_recv(msg_ch, RECV_SYNC1, &size_msg, &their_addr))>0)            //If queue is non-empty
	{
	  //printf("\nEntered apprecvclk with msg RECV_SYNC1");
	  tend = t_virt + tm_clock - t_init;                //Time at which cycle ends
	  mw = (msg_s *)msg_ch;
	  tsnd = mw->tm2;
	  trcv = mw->msw.tm1;
	  tstrt = mw->msw.m.tm0;
	  i = mw->msw.m.machine;
	  j = mw->msw.m.iteratn;
	  mw = NULL;
	  t_diff=(tend - tstrt + tsnd - trcv + 0.0L)/2.0;
	  if(t_diff>t_max[i])
	    t_max[i]=t_diff;
	  if(t_diff<t_min[i])
	    t_min[i]=t_diff;
	  t_avg[i]=(t_avg[i]*cycles[i] + t_diff)/(cycles[i]+1);
	  cycles[i]++;
          //printf("\nMachine:%d\tIteration:%lu\nMin: %llf, Avg: %llf, Max: %llf [Times denoted in msecs], Packets received: %ld out of %ld\n", i, j, t_min[i], t_avg[i], t_max[i], cycles[i], sent[i]);
	  if((tsnd + t_avg[i]) > tend)      //If the clock of the client is faster
	    tm_virt = tsnd + t_avg[i] + tm_clock - tm_init;
	  else                                //If clock of server is faster
	    tm_virt = tm_virt + tm_clock - tm_init;
	  tm_init = tm_clock;
	  tm_lastmsg[i] = tm_virt;
	  max_time=0.0;
	  for(k=0;k<hno;k++)
	    if(t_max[k]>max_time)
	      max_time = t_max[k];    //max_time gives the buffer time
	  
	  if( (((cycles[i]%10==0) && (i>hno)) ||(hno+1 == USERS)) && ((cycles_bully + 1)<=(cycles[i] / 10)) && electn==0)              // BULLY ALGORITHM. Modify to start only if leader fail is suspected.
	    {
	      cycles_bully++;
	      if((hno + 1)!=USERS)              //There may be nodes with a higher ID than this one
		{
		  max_wait_time=0.0;
		  for(k = hno+1;k<USERS;k++)
		    if(t_max[k]>max_wait_time)
		      max_wait_time = t_max[k]; //max_wait_time gives the time requested to wait
		  if(max_wait_time>max_time)
		    max_time = max_wait_time;
		  
		  bullyelectn *electnmsg, *ans_electnmsg;
		  electnmsg = (bullyelectn *)malloc(sizeof(bullyelectn));
		  electnmsg->type = MT_ELECTN_INIT;
		  electnmsg->iteratn = cycles_bully;
		  electnmsg->self_nodeno = hno;

		  for(k = hno+1; k<USERS; k++)  //Broadcasts a message of election
		    {
		      electnmsg->machine = k;
		      queue_send(electnmsg, sizeof(bullyelectn), addr[k]);
		    }
		  recvd = 0;
		  sleep((int)(max_wait_time + max_time));
		  usleep((max_wait_time+max_time-(int)max_wait_time-(int)max_time)*1000000L);
		  msg_ch = (char *)malloc(sizeof(bullyelectn)+sizeof(char));
		  while(queue_recv(msg_ch, MT_ANSWERMSG, &size_msg, &their_addr)>0)
		    {
		      ans_electnmsg = (bullyelectn *)msg_ch;
		      if(ans_electnmsg->iteratn == cycles_bully)
			{
			  recvd++;
			  break;
			}
		    }
		  free(electnmsg);
		  free(msg_ch);
		}
	      if((recvd==0)||((hno+1)==USERS))
		{
		  coord *coord_msg;
		  coord_msg = (coord *)malloc(sizeof(coord));
		  coord_msg->type = MT_COORD;
		  coord_msg->leader = hno;
		  for(k=0;k<hno;k++)
		    queue_send(coord_msg, sizeof(coord), addr[k]);
		  fprintf(f, "Node: %d says Leader- Node: %d,\tIP: %s\n", hno, hno, add[hno]);
		  free(coord_msg);
		}
	    }//END_OF_BULLYALGO
	}
      else
        usleep(1000);
    }
  return (NULL);
}
